package ds_industry_2025.industry.gy_08.T4

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{lit, to_timestamp}
import org.dom4j.DocumentHelper

import java.sql.Timestamp
import java.text.SimpleDateFormat
import java.util.{Calendar, Properties}
/*
    1、根据dwd库中fact_machine_data表（或MySQL的shtd_industry库中MachineData表），根据以下要求转换：获取最大分区
    （MySQL不用考虑）的数据后，首先解析列machine_record_data（MySQL中为MachineRecordData）的数据（数据格式为xml，采用
    dom4j解析，解析demo在客户端/home/ubuntu/Documents目录下），并获取每条数据的主轴转速，主轴倍率，主轴负载，进给倍率，进给速
    度，PMC程序号，循环时间，运行时间，有效轴数，总加工个数，已使用内存，未使用内存，可用程序量，注册程序量等相关的值（若该条数据没
    有相关值，则按下表设置默认值），同时转换machine_record_state字段的值，若值为报警，则填写1，否则填写0，以下为表结构，将数据
    保存在dwd.fact_machine_learning_data，使用hive cli按照machine_record_id升序排序，查询
    dwd.fact_machine_learning_data前1条数据，将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
    dwd.fact_machine_learning_data表结构：
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("特征")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")




    import spark.implicits._

    //  todo 读取mysql的数据
    val data = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_industry?useSSL=false", "MachineData", conn)


    val result = data.map(
      r => {
        //  todo 由于有一些字段的数据并不是依靠xml解析出来的，所以我们需要提前处理这些数据
        //  todo 获取row对象里面的一个字段的数据可以有两种方式 1.r(1).toString   2.r.getAs[String](1)
        val m = new machine()
        m.machine_record_id = r(0).toString.toInt
        m.machine_id = r(1).toString.toDouble
        m.machine_record_state = if (r.getAs[String](2).equals("报警")) 1.0 else 0.0
        //  parse 是 SimpleDateFormat 的方法，用于将字符串解析为 java.util.Date 对象。
        //  getTime 是 java.util.Date 的方法，返回自 1970 年 1 月 1 日 00:00:00 GMT 以来的毫秒数（即时间戳）。
        //  java.sql.Timestamp 是 java.util.Date 的子类，用于表示 SQL 中的时间戳类型。
        m.machine_record_date = new Timestamp(
          new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(r(4).toString).getTime
        )
        //   todo 接下来开始解析xml数据
        if (r.getAs[String](3) != null) {
          //  todo 由于dom4j解析dom的时候需要一个root节点，但是比赛的数据中没有，所以我们应该自己添加一个root节点
          //  下面使用<rows>作为root节点
          val machine_data = s"<rows>${r(3).toString}</rows>"
          //  todo 使用DocumentHelper.parseText(machine_data)将字符串解析为一个xml的document对象 ,document对象 可以操作xml文档
          val document = DocumentHelper.parseText(machine_data)
          //  获取 XML 文档的根节点（即 <rows>）。
          val root = document.getRootElement
          //  elementIterator() 获取根节点的子节点迭代器。
          val it = root.elementIterator()
          while (it.hasNext) {
            //  拿到当前行的数据
            val element = it.next()
            //  从当前子节点（element）中提取 ColName 属性的值，即字段名。
            val colName = element.attributeValue("ColName")
            //   // todo 如果得到的元素不是null，也不是空值
            if (!element.getTextTrim.equals("null") && element.getTextTrim.nonEmpty) {
              colName match {
                case "主轴转速" => m.machine_record_mainshaft_speed = element.getTextTrim.toDouble
                case "主轴倍率" => m.machine_record_mainshaft_multiplerate = element.getTextTrim.toDouble
                case "主轴负载" => m.machine_record_mainshaft_load = element.getTextTrim.toDouble
                case "进给倍率" => m.machine_record_feed_speed = element.getTextTrim.toDouble
                case "进给速度" => m.machine_record_feed_multiplerate = element.getTextTrim.toDouble
                case "PMC程序号" => m.machine_record_pmc_code = element.getTextTrim.toDouble
                case "循环时间" => m.machine_record_circle_time = element.getTextTrim.toDouble
                case "运行时间" => m.machine_record_run_time = element.getTextTrim.toDouble
                case "有效轴数" => m.machine_record_effective_shaft = element.getTextTrim.toDouble
                case "总加工个数" => m.machine_record_amount_process = element.getTextTrim.toDouble
                case "已使用内存" => m.machine_record_use_memory = element.getTextTrim.toDouble
                case "未使用内存" => m.machine_record_free_memory = element.getTextTrim.toDouble
                case "可用程序量" => m.machine_record_amount_use_code = element.getTextTrim.toDouble
                case "注册程序量" => m.machine_record_amount_free_code = element.getTextTrim.toDouble
                //  避免 MatchError 异常
                //如果 colName 的值没有匹配到任何 case 分支（例如 colName 是一个未定义的字段名），Scala 会抛出 MatchError 异常。
                case _ => ""
              }

            }
          }
        }
        // todo 将machine对象返回
        m
      }
    )

    result.show
    //  todo 把结果保存，给下面的随机森林使用
    result.write.format("hive").mode("overwrite")
      .saveAsTable("dwd.fact_machine_learning_data")

    spark.close()
  }
}
//  todo 由于字段较多，下面设置一个样例类，还可以设置默认值
case class machine(
                    var machine_record_id:Int= 0,
                    var machine_id:Double = 0.0,
                    var machine_record_state:Double = 0.0,
                    var machine_record_mainshaft_speed:Double = 0.0,
                    var machine_record_mainshaft_multiplerate:Double = 0.0,
                    var machine_record_mainshaft_load:Double =0.0,
                    var machine_record_feed_speed:Double=0.0,
                    var machine_record_feed_multiplerate:Double=0.0,
                    var machine_record_pmc_code:Double=0.0,
                    var machine_record_circle_time:Double=0.0,
                    var machine_record_run_time:Double=0.0,
                    var machine_record_effective_shaft:Double=0.0,
                    var machine_record_amount_process:Double=0.0,
                    var machine_record_use_memory:Double=0.0,
                    var machine_record_free_memory:Double=0.0,
                    var machine_record_amount_use_code:Double=0.0,
                    var machine_record_amount_free_code:Double=0.0,
                    var machine_record_date:Timestamp=null,
                    var dwd_insert_user:String="user1",
                    var dwd_insert_time:Timestamp=new Timestamp(System.currentTimeMillis()),
                    var dwd_modify_user:String="user1",
                    var dwd_modify_time:Timestamp=new Timestamp(System.currentTimeMillis())

                  )
